package com.acmedcare.framework.exchange.center.core;

import com.acmedcare.framework.exchange.center.core.exception.PluginNotFoundException;
import com.acmedcare.framework.exchange.center.core.handler.AgentHandlerMBean;
import com.acmedcare.framework.exchange.center.core.handler.ConsumerHandlerMBean;
import com.acmedcare.framework.exchange.center.core.handler.PluginHandlerMBean;
import com.acmedcare.framework.exchange.center.entity.bean.Plugin;
import com.acmedcare.framework.exchange.center.entity.dto.Hdfs;
import java.io.IOException;
import java.lang.reflect.Method;
import java.net.ServerSocket;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.management.MBeanServerConnection;
import javax.management.MBeanServerInvocationHandler;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cglib.proxy.MethodInterceptor;
import org.springframework.cglib.proxy.MethodProxy;

@Slf4j
public class PluginContainerMBeanProxy implements MethodInterceptor {

  final String bank = " ";
  final String serverPath;
  final ThreadPoolExecutor pool;
  final Hdfs hdfs;

  public PluginContainerMBeanProxy(Hdfs hdfs) {
    this.hdfs = hdfs;
    pool = new ThreadPoolExecutor(10, 30, 2,
        TimeUnit.MINUTES, new ArrayBlockingQueue<>(10));
    serverPath = hdfs.getServerPath();
  }

  @Override
  public Object intercept(Object o, Method method, Object[] objects, MethodProxy methodProxy)
      throws Throwable {
    //使用动态代理，透明化路由进程之间的交互
    Plugin plugin;
    switch (method.getName()) {
      case "init":
        List<Plugin> plugins = (List<Plugin>) objects[0];
        plugins.forEach(p -> {
          try {

            PluginContainerMBean mBean = getMBean(p.getName());
            List<Plugin> pluginList = new ArrayList<>();
            pluginList.add(p);
            mBean.init(pluginList);
          } catch (PluginNotFoundException e) {
            log.info(String.format("插件:%s初始化失败：%s", p.getName(), e.getMessage()));
          }
        });
        break;
      case "unloadAsync":
        plugin = (Plugin) objects[0];
        getMBean(plugin.getName()).unloadAsync(plugin);
        break;
      case "load":
        plugin = (Plugin) objects[0];
        getMBean(plugin.getName()).load(plugin);
        break;
      case "getPlugin":
        String pluginName = objects[0].toString();
        return getMBean(pluginName).getPlugin(pluginName);
      case "getPlugins":
        List<Plugin> pluginList = new ArrayList<>();
        JmxContainerUtil.GroupContainer.values()
            .forEach(p -> p.getPlugins().forEach(pp -> pluginList.add(pp)));
        return pluginList;
      case "restart":
        pluginName = objects[0].toString();
        reset(pluginName);
        break;
      case "setEnv":
        JmxContainerUtil.GroupContainer.values()
            .forEach(c -> c.setEnv(objects[0].toString(), objects[1].toString()));
    }

    return null;
  }

  private PluginContainerMBean getMBean(String pluginName) throws PluginNotFoundException {

    pluginName = pluginName.toLowerCase();
    if (JmxContainerUtil.GroupContainer.get(pluginName) == null) {
      try {

        PluginContainerMBean containerMBean = createMBeanContainer(pluginName);
        JmxContainerUtil.GroupContainer.put(pluginName, containerMBean);
      } catch (Exception e) {
        log.error(String.format("插件:%s，初始化失败!", pluginName), e);
        //初始化失败，put一个空的容器，防止失败后重复创建
        JmxContainerUtil.GroupContainer.put(pluginName, new ErrorPluginContainer());
      }
    }

    PluginContainerMBean mBean = JmxContainerUtil.GroupContainer.get(pluginName);
    if (mBean instanceof ErrorPluginContainer) {
      throw new PluginNotFoundException(String.format("插件:%s,加载失败!请尝试重新加载,或者重启插件。", pluginName));
    }

    return mBean;
  }

  private void reset(String pluginName) {

    pluginName = pluginName.toLowerCase();
    try {
      PluginContainerMBean containerMBean = createMBeanContainer(pluginName);
      JmxContainerUtil.GroupContainer.put(pluginName, containerMBean);
    } catch (Exception e) {
      log.error(String.format("插件:%s,重启失败", pluginName), e);
      JmxContainerUtil.GroupContainer.put(pluginName, new ErrorPluginContainer());
    }
  }

  private PluginContainerMBean createMBeanContainer(String pluginName)
      throws Exception {

    PluginProcessMetadata processMetadata = new PluginProcessMetadata();
    processMetadata.setPluginName(pluginName);
    if (JmxContainerUtil.Processes.get(pluginName) == null) {

      processMetadata.setPort(getFreePort());
      Process process = createMBeanServer(processMetadata.getPluginName(),
          processMetadata.getPort());
      processMetadata.setProcess(process);

    } else {

      processMetadata = JmxContainerUtil.Processes.get(pluginName);
      if (processMetadata.getProcess().isAlive()) {
        //如果进程不存活则直接销毁重新创建新的进程
        processMetadata.getProcess().destroyForcibly();
      }

      Process process = createMBeanServer(processMetadata.getPluginName(),
          processMetadata.getPort());
      processMetadata.setProcess(process);
    }
    JmxContainerUtil.Processes.put(pluginName, processMetadata);

    return createJmxConnection(pluginName, processMetadata, 8);
  }

  private PluginContainerMBean createJmxConnection(String pluginName,
      PluginProcessMetadata processMetadata, int retry) throws PluginNotFoundException {

    if (retry <= 0) {
      log.info("插件链接创建失败,并且已经达到最大重试次数,请重启插件后再试!");

      throw new PluginNotFoundException(String.format("插件:%s,加载失败!请尝试重新加载,或者重启插件。", pluginName));
    }
    try {

      //等待5秒后创建桥接端链接
      Thread.sleep(5 * 1000L);
      String uri = String
          .format("service:jmx:rmi:///jndi/rmi://127.0.0.1:%s/jmxrmi", processMetadata.getPort());
      JMXServiceURL url = new JMXServiceURL(uri);
      JMXConnector jmxConnector = JMXConnectorFactory.connect(url, null);
      MBeanServerConnection serverConnection = jmxConnector.getMBeanServerConnection();
      ObjectName mbeanName = new ObjectName("jmxBean:name=pluginContainer");

      log.info(String.format("插件：%s，桥接端正在创建链接......", pluginName));
      PluginContainerMBean proxy = MBeanServerInvocationHandler
          .newProxyInstance(serverConnection, mbeanName, PluginContainerMBean.class, false);

      JmxContainerUtil.AgentHandlers.put(pluginName, MBeanServerInvocationHandler
          .newProxyInstance(serverConnection, new ObjectName("jmxBean:name=agentHandler"),
              AgentHandlerMBean.class, false));

      JmxContainerUtil.ConsumerHandlers.put(pluginName, MBeanServerInvocationHandler
          .newProxyInstance(serverConnection, new ObjectName("jmxBean:name=consumerHandler"),
              ConsumerHandlerMBean.class, false));

      JmxContainerUtil.PluginHandlers.put(pluginName, MBeanServerInvocationHandler
          .newProxyInstance(serverConnection, new ObjectName("jmxBean:name=pluginHandler"),
              PluginHandlerMBean.class, false));
      log.info(String.format("插件：%s，桥接链接创建完毕......", pluginName));
      return proxy;
    } catch (Exception e) {
      retry--;
      log.info(String
          .format("插件：%s，桥接链接创建失败,正在进行重试,剩余重试次数%s,错误原因:%s", pluginName, retry, e.getMessage()));
      return createJmxConnection(pluginName, processMetadata, retry);
    }
  }

  private Process createMBeanServer(String pluginName, Integer port) throws IOException {

    StringBuilder builder = new StringBuilder();

    builder.append("java");
    builder.append(bank);
    // jvm memory
    builder.append(
        "-server -Xms28m -Xmx128m -Xmn28m -XX:MetaspaceSize=28m -XX:MaxMetaspaceSize=256m");
    builder.append(bank);
    // jvm gc
    builder.append(
        "-XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8  -XX:-UseParNewGC");
    //设置dump文件路径
    builder.append(bank);

    //打印gc并且在crash之后设置dump文件
    builder.append(String.format(
        "-verbose:gc -Xloggc:%s_gc.log -XX:HeapDumpPath=%s/HeapDumpOnOutOfMemoryError/ -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime -XX:+PrintAdaptiveSizePolicy"
        , pluginName, pluginName));

    builder.append(bank);
    builder.append("-classpath");
    builder.append(bank);
    builder.append(serverPath);
    builder.append(bank);
    builder.append("com.acmedcare.framework.exchange.center.jmx.JmxServer");
    builder.append(bank);
    builder.append(pluginName);
    builder.append(bank);
    builder.append(port);
    builder.append(bank);
    builder.append(JmxContainerUtil.ServerPort);
    Runtime runtime = Runtime.getRuntime();
    Process process = runtime.exec(builder.toString());
    log.info(String.format("插件：%s，正在启动，启动脚本：%s", pluginName, builder.toString()));

    pool.submit(() -> StreamReaderUtil.readToFile(process.getInputStream()));
    pool.submit(() -> StreamReaderUtil.readToFile(process.getErrorStream()));
    return process;
  }

  private Integer getFreePort() throws IOException {

    ServerSocket serverSocket = new ServerSocket(0); //读取空闲的可用端口
    Integer port = serverSocket.getLocalPort();
    serverSocket.close();
    return port;
  }
}
